package org.apache.spark.streaming.kafka

/**
  * streaming.kafka 工具类 本身是私有的取相同包名
  * wzxjava@126.com
  * Created by wangzhixuan on 2017/05/26 16:47
  */
class KafkaManager(val kafkaParams: Map[String, String]) {

  private val kc = new KafkaCluster(kafkaParams)

  /**
    * 获取kafka中offset的最小值
    *
    * @param topic
    * @return
    */
  def getEarliestLeaderOffsets(topic: String): Map[Int, Long] = {
    val partitions = kc.getPartitions(Set(topic)).right.get
    val partitionToOffset = kc.getEarliestLeaderOffsets(partitions).right.get
    val offsets = partitionToOffset.map(x => (x._1.partition, x._2.offset))
    offsets
  }
}
